package com.wenge.datagroup.storage.util

import org.apache.commons.lang3.StringUtils
import redis.clients.jedis.{HostAndPort, JedisCluster, JedisPoolConfig}

import java.util.{HashSet => JavaHashSet}

/**
 * @author
 *   Depp
 * @description
 * @date
 *   2022-09-02 19:53
 */
class RedisSink(makeJedisCluster: () => JedisCluster) {
  lazy val jedisCluster: JedisCluster = makeJedisCluster()
}

object RedisSink {
  def apply(hostAndPort: String, password: String): RedisSink = {
    val createJedisPoolFunc = () => {
      val hostAndPorts = new JavaHashSet[HostAndPort]()
      val hostsAndPortsArr: Array[String] = hostAndPort.split(",")
      // 增加redis密码配置
      for (hostAndPort <- hostsAndPortsArr) {
        val hostAndPortElem: Array[String] = hostAndPort.split(":")
        hostAndPorts.add(new HostAndPort(hostAndPortElem(0), hostAndPortElem(1).toInt))
      }

      var jedisCluster: JedisCluster = null
      if (StringUtils.isNotBlank(password)) {
        jedisCluster =
          new JedisCluster(hostAndPorts, 10000, 10000, 10, password, getJedisPoolConfig)
      } else {
        jedisCluster = new JedisCluster(hostAndPorts, 10000, 10000, 10, getJedisPoolConfig)
      }
      val hook = new Thread {
        override def run = {
          jedisCluster.close()
        }
      }
      sys.addShutdownHook(hook.run)
      jedisCluster
    }
    new RedisSink(createJedisPoolFunc)
  }

  private[this] def getJedisPoolConfig: JedisPoolConfig = {
    val jedisPoolConfig = new JedisPoolConfig()
    jedisPoolConfig.setMaxTotal(100) // 最大连接数
    jedisPoolConfig.setMaxIdle(20) // 最大空闲
    jedisPoolConfig.setMinIdle(20) // 最小空闲
    jedisPoolConfig.setBlockWhenExhausted(true) // 忙碌时是否等待
    jedisPoolConfig.setMaxWaitMillis(50000) // 忙碌时等待时长 毫秒
    jedisPoolConfig.setTestOnBorrow(true) // 每次或得链接进行测试
    jedisPoolConfig
  }
}
